package com.qqt.csr.archive.service.listener;

import com.qqt.csr.archive.batch.task.ArchiveRunner;
import com.qqt.csr.archive.entity.reader.ArtificialConversation;
import com.qqt.csr.archive.entity.reader.SmartConversation;
import com.qqt.csr.archive.entity.writer.ArchiveConversation;
import com.qqt.csr.archive.model.ConversationEndMessage;
import com.qqt.csr.archive.service.ArchiveConversationService;
import com.qqt.csr.archive.service.ArtificialConversationService;
import com.qqt.csr.archive.service.SmartConversationService;
import com.qqt.csr.common.utils.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * @Author: yang hao
 * @Date: 2024/11/19 10:08
 * @Description: 会话结束事件监听
 **/
@Slf4j
@Service
@SuppressWarnings("unused")
@RocketMQMessageListener(
        consumerGroup = "${rocketmq.consumer.group}" + "_" + "conversation_end_event",
        topic = "archive_event_topic",
        selectorType = SelectorType.TAG,
        selectorExpression = "conversation_end_event",
        messageModel = MessageModel.CLUSTERING,
        tlsEnable="${rocketmq.consumer.tlsEnable}"
)
public class ConversationEndEventListener implements RocketMQListener<ConversationEndMessage> {

    @Autowired
    private ArtificialConversationService artificialConversationService;
    @Autowired
    private SmartConversationService smartConversationService;
    @Autowired
    private ArchiveConversationService archiveConversationService;

    @Autowired
    private ArchiveRunner archiveRunner;

    @Override
    public void onMessage(ConversationEndMessage message) {
        // 校验会话结束通知消息
        if (!check(message)) {
            log.error("会话结束消息校验不通过，消息:{}", JsonUtil.toJson(message));
            return;
        }
        // 校验会话是否已归档
        List<String> existConversationIds = isExist(message.getConversationIds());
        if (existConversationIds.isEmpty()) {
            log.info("会话已归档，message:{}", JsonUtil.toJson(message));
            return;
        }
        message.setConversationIds(existConversationIds);
        archiveRunner.run(message);
    }

    /**
     * 校验会话结束消息
     * @param message {@link ConversationEndMessage} 会话结束消息
     * @return {@link Boolean} 校验结果
     */
    private boolean check(ConversationEndMessage message) {
        // 校验会话ID列表
        if (message.getConversationIds() == null || message.getConversationIds().isEmpty()) {
            log.error("会话结束消息中，conversationIds为空，message:{}", JsonUtil.toJson(message));
            return false;
        }
        // 校验来源类型
        switch (message.getSource()) {
            case ARTIFICIAL -> {
                // 查询人工会话
                List<ArtificialConversation> artificialConversationList = artificialConversationService.getByIds(message.getConversationIds());
                // 校验会话列表是否为空 && 校验数量是否一致
                return validateConversations(artificialConversationList, message.getConversationIds());
            }
            case SMART -> {
                // 查询智能会话
                List<SmartConversation> smartConversationList = smartConversationService.getByIds(message.getConversationIds());
                // 校验会话列表是否为空 && 校验数量是否一致
                return validateConversations(smartConversationList, message.getConversationIds());
            }
        }
        log.error("会话结束消息，source未匹配成功，message:{}", JsonUtil.toJson(message));
    	return false;
    }

    /**
     * 校验会话是否已归档
     * @param conversationIds {@link List<String>} 会话ID列表
     * @return {@link List<String>} 不存在的会话id
     */
    private List<String> isExist(List<String> conversationIds) {
        List<ArchiveConversation> archiveConversations = archiveConversationService.getByConversationIds(conversationIds);
        List<String> archiveConversationIds = archiveConversations.stream().map(ArchiveConversation::getConversationId).toList();
        return conversationIds.stream().filter(id -> !archiveConversationIds.contains(id)).toList();
    }

    /**
     * 校验会话列表
     * @param conversationList {@link List} 会话列表
     * @param conversationIds {@link List<String>} 会话ID列表
     * @return {@link Boolean} 校验会话列表是否为空 && 校验数量是否一致
     */
    private boolean validateConversations(List<?> conversationList, List<String> conversationIds) {
        return conversationList != null
                && !conversationList.isEmpty()
                && conversationList.size() == conversationIds.size();
    }

}
